package com.binance.zkconfig.app.service.impl;

import com.binance.zkconfig.app.model.ZkConfigChange;
import com.binance.zkconfig.app.service.ZkConfig;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * Created by Fei.Huang on 2018/8/28.
 */
@RequiredArgsConstructor
public class ZkConfigImpl implements ZkConfig {

    @Getter
    private final String namespace;
    private Map<String, String> properties = Maps.newConcurrentMap();

    @Override
    public Set<String> getPropertyNames() {
        return this.properties.keySet();
    }

    @Override
    public String getProperty(String key) {
        return this.properties.get(key);
    }

    @Override
    public void put(String key, String value) {
        this.properties.put(key, value);
    }

    @Override
    public void remove(String key) {
        this.properties.remove(key);
    }

    @Override
    public synchronized void syncProperties(Map<String, String> properties) {
        List<ZkConfigChange> zkConfigChanges = calcPropertyChanges(properties);
        if (!CollectionUtils.isEmpty(zkConfigChanges)) {
            for (ZkConfigChange zkConfigChange : zkConfigChanges) {
                ZkConfigChange.Type type = zkConfigChange.getType();
                if (ZkConfigChange.Type.ADD.equals(type)) {
                    this.properties.put(zkConfigChange.getKey(), zkConfigChange.getNewValue());
                } else if (ZkConfigChange.Type.REMOVE.equals(type)) {
                    this.properties.remove(zkConfigChange.getKey());
                } else if (ZkConfigChange.Type.UPDATE.equals(type)) {
                    this.properties.put(zkConfigChange.getKey(), zkConfigChange.getNewValue());
                }
            }
        }
    }

    private List<ZkConfigChange> calcPropertyChanges(Map<String, String> newProperties) {
        List<ZkConfigChange> zkConfigChanges = new ArrayList<>();
        Set<String> previousKeys = properties.keySet();
        Set<String> currentKeys = newProperties.keySet();
        Set<String> commonKeys = Sets.intersection(previousKeys, currentKeys);
        Set<String> newKeys = Sets.difference(currentKeys, commonKeys);
        Set<String> removedKeys = Sets.difference(previousKeys, commonKeys);
        for (String key : newKeys) {
            zkConfigChanges.add(ZkConfigChange.builder().type(ZkConfigChange.Type.ADD).key(key)
                    .newValue(newProperties.get(key)).build());
        }
        for (String key : removedKeys) {
            zkConfigChanges.add(ZkConfigChange.builder().type(ZkConfigChange.Type.REMOVE).key(key)
                    .oldValue(properties.get(key)).build());
        }
        for (String key : commonKeys) {
            String previousValue = properties.get(key);
            String currentValue = newProperties.get(key);
            if (!Objects.equals(previousValue, currentValue)) {
                zkConfigChanges.add(ZkConfigChange.builder().type(ZkConfigChange.Type.UPDATE).key(key)
                        .oldValue(previousValue).newValue(currentValue).build());
            }
        }
        return zkConfigChanges;
    }
}